package com.gandong8.trace.tool.consumer.ons;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.gandong8.trace.tool.util.JsonUtil;
import com.gandong8.trace.tool.util.TraceUtil;
import lombok.extern.slf4j.Slf4j;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;

@Slf4j
public abstract class AbstractOnsListener<T> implements MessageListener {
    private Class<T> recordClazz;

    public AbstractOnsListener() {
        parseClazz();
    }

    @SuppressWarnings("unchecked")
    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        TraceUtil.setLogTrace(message.getUserProperties(TraceUtil.TRACE_TAG));
        String topic = message.getTopic();
        String tags = message.getTag();
        String body = new String(message.getBody(), StandardCharsets.UTF_8);
        log.info("收到消息topic={}, tags={}, 消息内容={}", topic, tags, body);
        try {
            if ("String".equals(recordClazz.getSimpleName())) {
                execute((T) body);
            } else {
                T obj = JsonUtil.toObj(body, recordClazz);
                execute(obj);
            }
        } catch (Exception e) {
            log.error("消息处理异常", e);
        }
        return Action.CommitMessage;
    }



    protected abstract void execute(T message);



    @SuppressWarnings("unchecked")
    private void parseClazz() {
        Type type = getClass().getGenericSuperclass();
        if (type instanceof ParameterizedType) {
            recordClazz = (Class<T>) ((ParameterizedType) type).getActualTypeArguments()[0];
        }
    }



}
